package com.lmh.service.impl;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson2.JSONObject;
import com.lmh.bean.JobState;
import com.lmh.bean.OfflineAnalyticTemplate;
import com.lmh.config.TaskConfig;
import com.lmh.job.OfflineAnalyticJob;
import com.lmh.mapper.OfflineAnalyticMapper;
import com.lmh.service.OfflineAnalyticService;

import com.lmh.utils.DataBasePoolUtil;
import com.lmh.utils.JedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

/**
 * @program bigdata-service
 * @Description 读取配置获取对应sql
 * @Author limeihua
 * @Date 2024/08/14 19:03
 * @Version 1.0
 */
@Service
@Slf4j
public class OfflineAnalyticServiceImpl implements OfflineAnalyticService {

    @Autowired
    private OfflineAnalyticMapper offlineAnalyticMapper;


    @Override
    public String execute(JSONObject params) {
        String sqlID = params.getString("sql_id");
        OfflineAnalyticTemplate offlineAnalyticTemplate = offlineAnalyticMapper.getSqlContent(sqlID);
        String executionParams = offlineAnalyticTemplate.getExecutionParams();
        JSONObject executionJson = JSONObject.parseObject(executionParams);
        String taskType = executionJson.getString("task_type");
        if("offline".equals(taskType)){
            return asynchronousExecute(params);
        } else if ("realtime".equals(taskType)) {
            // 实时任务处理逻辑
            DruidPooledConnection connection = null;
            String sqlContent = offlineAnalyticTemplate.getSqlContent();
            String databaseType = executionJson.getString("databaseType");
            log.info("即将执行实时分析sql：\n【" + sqlContent + "】");
            DruidDataSource dataSource = DataBasePoolUtil.getDataSource(databaseType);
            try {
                connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(sqlContent);
                while (resultSet.next()){
//                装配返回参数
                    return resultSet.getString(1);
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }finally {
                if(connection != null){
                    try {
                        connection.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
            return null;
        }else {
            throw new IllegalArgumentException("任务类型错误");
        }
    }

    @Override
    public String asynchronousExecute(JSONObject params) {
        String result = "";
        Jedis jedis = JedisUtil.getJedis();
        // redis里面的key用sql_id加上查询条件
        String[] sqlIDs = params.getString("sql_id").split(",");
        List<OfflineAnalyticJob> jobs = new ArrayList<>();
        for (String sqlID : sqlIDs) {
            String value = jedis.get("");
            if(value != null) continue;
            OfflineAnalyticTemplate offlineAnalyticTemplate = offlineAnalyticMapper.getSqlContent(sqlID);
            OfflineAnalyticJob job = new OfflineAnalyticJob(offlineAnalyticTemplate);
            jobs.add(job);
            TaskConfig.addTask(job);
        }
        while (jobs.size() != 0){
            for (OfflineAnalyticJob job : jobs) {
                JobState jobStatus = job.getJobStatus();
                if(jobStatus == JobState.FINISH){
                    jobs.remove(job);
                } else if (jobStatus == JobState.DIED) {
                    throw new RuntimeException("任务执行失败！！！");
                }
            }
        }
        for (String sqlID : sqlIDs) {
            result += jedis.get("");
        }
        return result;
    }
}
